package com.depository.oss.quere;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisConnectionUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author xiaomei
 * @version V1.0
 * @Title: RedisQueue
 * @Package com.xiaomei.queue.redis.demo01
 * @Description:
 * @date 11/7/17
 */
public class RedisQueue<T> implements InitializingBean,DisposableBean,IRedisQueue<T> {

    //队列名称
    public String queueName;
    //原生key
    private byte[] rawKey;

    private RedisTemplate redisTemplate;
    private RedisConnectionFactory factory;
    private RedisConnection connection; //为了堵塞
    private BoundListOperations<String, T> listOperations;
    private Lock lock = new ReentrantLock();//基于底层IO阻塞考虑  如果分布式的话，就是用分式式的锁





    @Override
    public T take() throws InterruptedException {
        return poll(0);
    }

    @Override
    public T takeOpposite() throws InterruptedException {
        return pollOpposite(0);
    }

    @Override
    public T poll(int seconds) throws InterruptedException {
        lock.lockInterruptibly();
        try{
            List<byte[]> results = connection.bRPop(seconds, rawKey);
            if(CollectionUtils.isEmpty(results)){
                return null;
            }
            return (T)redisTemplate.getValueSerializer().deserialize(results.get(1));
        }finally{
            lock.unlock();
        }
    }

    @Override
    public T pollOpposite(int seconds) throws InterruptedException {
        lock.lockInterruptibly();
        try{
            List<byte[]> results = connection.bLPop(seconds, rawKey);
            if(CollectionUtils.isEmpty(results)){
                return null;
            }
            return (T)redisTemplate.getValueSerializer().deserialize(results.get(1));
        }finally{
            lock.unlock();
        }
    }

    @Override
    public void add(T value) {
        listOperations.rightPush(value);
    }

    @Override
    public void addOpposite(T value) {
        listOperations.leftPush(value);
    }

    @Override
    public void clearAll() {
//        listOperations.
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        factory = redisTemplate.getConnectionFactory();
        connection = RedisConnectionUtils.getConnection(factory);
        rawKey = redisTemplate.getKeySerializer().serialize(queueName);
        listOperations = redisTemplate.boundListOps(queueName);
    }


    @Override
    public void destroy() throws Exception {
        RedisConnectionUtils.releaseConnection(connection, factory);
    }



    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }

    public void setRedisTemplate(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }


}
